package com.shujia.flink.core

import org.apache.flink.api.common.state.{ListStateDescriptor, MapStateDescriptor}
import org.apache.flink.api.common.state.ListState
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.{BroadcastProcessFunction, CoMapFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

object Demo10Connect {
  def main(args: Array[String]): Unit = {
    /**
      * 动态拦截黑名单
      *
      */

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val ds = env.socketTextStream("node1", 6666)

    val ds1 = env.socketTextStream("node1", 7777)

    /**
      * union  合并两个流   两个流的数据类型必须一致
      *
      */

    //val unionDS = ds.union(ds1)

    val connectDS: ConnectedStreams[String, String] = ds.connect(ds1)

    val ds3 = connectDS.map(new CoMapFunction[String, String, (String, Int)] {

      //处理ds中的数据
      override def map1(value: String): (String, Int) = {
        (value, 1)
      }

      //处理ds1中的数据
      override def map2(value: String): (String, Int) = {
        (value, 1)
      }
    })

    ds3.keyBy(_._1)
      .sum(1)
      .print()


    env.execute()
  }
}


